//package com.jiegui.common.kafka;
//
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.support.SendResult;
//import org.springframework.stereotype.Component;
//import org.springframework.util.concurrent.ListenableFuture;
//import org.springframework.util.concurrent.ListenableFutureCallback;
//
//import javax.annotation.Resource;
//
//
///**
// * 发送kafka消息
// **/
//@Component
//@Slf4j
//@ConditionalOnProperty(name="spring.kafka.producer.retries")
//public class KafkaMessageProducer {
//
//    @Value("${spring.profiles.active}")
//    String tag;
//
//    @Resource
//    KafkaTemplate<String, String> kafkaTemplate;
//
//    @Value("${kafka.kafkaUse}")
//    private int kafkaUse;
//    /**
//     * 发送消息
//     *
//     * @param message kafka消息
//     */
//    public void sendMessage(KafkaMessage message) {
//            sendMessage(message, false);
//
//    }
//
//    /**
//     * 发送消息
//     *
//     * @param message kafka消息
//     * @param silent silent 是否不打印请求参数
//     */
//    public void sendMessage(KafkaMessage message, boolean silent,int...repeatCount) {
//        if(kafkaUse == 0){
//
//            return;
//        }
//        if (!silent) {
//            log.debug("kafka 请求参数:[{}]", message);
//        }
//        try {
//            ListenableFuture<SendResult<String, String>> future = null;
//            if (StringUtils.isNotBlank(message.getKey())) {
//                future = kafkaTemplate.send(message.getTopic(), message.getKey(), message.getBody());
//            } else {
//                future = kafkaTemplate.send(message.getTopic(), message.getBody());
//            }
//            // 发送回调，异常就执行 和 抛异常执行3次   2选1
//            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
//                @Override
//                public void onFailure(Throwable ex) {
//                    log.error("发送kafka异常onFailure: [{}]", ex.getMessage());
//
//                }
//                @Override
//                public void onSuccess(SendResult<String, String> result) {
//                    log.debug("kafka 发送成功:[{}]", message);
//                }
//            });
//
//        } catch (Exception e) {
//            log.error("发送kafka异常: [{}]", e.getMessage());
//            e.printStackTrace();
//
//    }
//
////
//
//    }
//
//
//}
